package ckR

import (
	cp "db2s/componentInit"
	mq "db2s/topic-mq"
	"reflect"
	"strings"
)

func MQCountListenersTableFunc(t, p int) map[Topic]*cp.MQTopicListeners {
	return map[Topic]*cp.MQTopicListeners{
		//有索引表的行校验
		IsMode:     cp.NewMQTopicListeners(cp.Topic(IsMode), t, 1, isModeListeners, 5, 10),
		CountPoint: cp.NewMQTopicListeners(cp.Topic(CountPoint), t, 1, countPointExecListeners, 5, 10),
	}
}
func MQRowsListenersTableFunc(t, p int) map[Topic]*cp.MQTopicListeners {
	return map[Topic]*cp.MQTopicListeners{
		//有索引表的行校验
		//IsSegmentQuery:       cp.NewMQTopicListeners(cp.Topic(IsSegmentQuery), t, 1, rowsIsSegmentQueryListeners, 5, 10),
		//IndexConcurrentQuery: cp.NewMQTopicListeners(cp.Topic(IndexConcurrentQuery), t*p, 1, rowsIndexConcurrentQueryListeners, 5, 20),
		//QueryTableSql:        cp.NewMQTopicListeners(cp.Topic(QueryTableSql), t*p, 200, rowsQueryTableSqlListeners, 1, 30),
		//QueryTableData:       cp.NewMQTopicListeners(cp.Topic(QueryTableData), t*p, 200, rowsQueryTableDataSqlListeners, 5, 41),
		//Md5Rows:              cp.NewMQTopicListeners(cp.Topic(Md5Rows), t*p, 200, rowsMd5Listeners, 5, 42),
		//Arrcmp:               cp.NewMQTopicListeners(cp.Topic(Arrcmp), t*p, 200, rowsArrcmpListeners, 5, 43),
		//AbnormalDataDiff:     cp.NewMQTopicListeners(cp.Topic(AbnormalDataDiff), t*p, 200, rowsAbnormalDataDiffListeners, 5, 50),
		//DataFixDiff:          cp.NewMQTopicListeners(cp.Topic(DataFixDiff), t*p, 200, rowsDataFixDiffListeners, 5, 60),
	}
}
func MQSampListenersTableFunc(t, p int) map[Topic]*cp.MQTopicListeners {
	return map[Topic]*cp.MQTopicListeners{
		//有索引表的行校验
		//IsSegmentQuery:       cp.NewMQTopicListeners(cp.Topic(IsSegmentQuery), t, 1, rowsIsSegmentQueryListeners, 5, 10),
		//IndexConcurrentQuery: cp.NewMQTopicListeners(cp.Topic(IndexConcurrentQuery), t*p, 1, rowsIndexConcurrentQueryListeners, 5, 20),
		//QueryTableSql:        cp.NewMQTopicListeners(cp.Topic(QueryTableSql), t*p, 200, rowsQueryTableSqlListeners, 1, 30),
		//QueryTableData:       cp.NewMQTopicListeners(cp.Topic(QueryTableData), t*p, 200, rowsQueryTableDataSqlListeners, 5, 41),
		//Md5Rows:              cp.NewMQTopicListeners(cp.Topic(Md5Rows), t*p, 200, rowsMd5Listeners, 5, 42),
		//Arrcmp:               cp.NewMQTopicListeners(cp.Topic(Arrcmp), t*p, 200, rowsArrcmpListeners, 5, 43),
		//AbnormalDataDiff:     cp.NewMQTopicListeners(cp.Topic(AbnormalDataDiff), t*p, 200, rowsAbnormalDataDiffListeners, 5, 50),
		//DataFixDiff:          cp.NewMQTopicListeners(cp.Topic(DataFixDiff), t*p, 200, rowsDataFixDiffListeners, 5, 60),
	}
}
func MQListenersTableSlice(t, p int) (res []*cp.MQTopicListeners) {
	for _, v := range []map[Topic]*cp.MQTopicListeners{
		MQCountListenersTableFunc(t, p),
	} {
		for _, vv := range v {
			res = append(res, vv)
		}
	}
	return
}
func rowsMQListenInitFunc(t, p int) map[cp.CheckMode][]*cp.MQTopicListeners {
	return map[cp.CheckMode][]*cp.MQTopicListeners{
		"rows": MQListenersTableSlice(t, p),
	}
}

func topicCreateInit(checkMod string, t, p int) (m map[cp.CheckMode][]*cp.MQTopicListeners) {
	m = make(map[cp.CheckMode][]*cp.MQTopicListeners)
	if funcName, ok := MqListenerSInitFunc[checkMod]; ok {
		if funcName.Kind() == reflect.Func {
			m1 := funcName.Call([]reflect.Value{
				reflect.ValueOf(t),
				reflect.ValueOf(p),
			})
			for i := 0; i < len(m1); i++ {
				m = m1[i].Interface().(map[cp.CheckMode][]*cp.MQTopicListeners)
			}
		}
	}
	return m
}
func GetCheckToTopicMQListeners() map[cp.CheckMode][]*cp.MQTopicListeners {
	var (
		t = GlobalPConfigs.rules.Mtc
		r = GlobalPConfigs.rules.Mcc
	)
	return topicCreateInit(GlobalPConfigs.rules.CheckMode, t, r)
}

func (context *SchedulePlanContext) topicEndAdd(index CheckMode, topic Topic) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndAdd(cp.Topic(topic))
}
func (context *SchedulePlanContext) topicEndMonit(index CheckMode, topic Topic) int64 {
	return context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndMonit(cp.Topic(topic))
}
func (context *SchedulePlanContext) topicEndDelete(index CheckMode, topic Topic) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndDelete(cp.Topic(topic))
}

func (context *SchedulePlanContext) topicEndDeleteNumber(index CheckMode, topic Topic, n int64) {
	context.AutoCloseMqComponentTable[cp.CheckMode(index)].TopicEndDeleteNumber(cp.Topic(topic), n)
}

func sendMsg(check CheckMode, topic Topic, tableName string, msg map[string]any) bool {
	var outFay string
	PlanContext.topicEndAdd(check, topic)
	if err := producer.Send(mq.Message{
		Topic:      string(topic),
		Body:       []byte(tableName),
		Properties: msg,
	}); err != nil {
		return false
	}
	if v, ok := msg["outFay"]; ok {
		outFay = v.(string)
	}
	if strings.EqualFold(outFay, "bar") {
		PlanContext.taskBarEndTopicAdd(msg)
	}
	return true
}

type FatherSon struct {
	FatherTags    map[string]any
	FatherSonTags map[string][]string
}
type SendMsgInfo struct {
	CheckMod  CheckMode
	Topic     Topic
	TableName string
}

func deleteTopic(topicMsg map[string]any) {
	var (
		dTopic Topic
		outFay = "bar"
	)
	if v, ok := topicMsg["dTopic"]; ok {
		dTopic = v.(Topic)
	}
	if v, ok := topicMsg["outFay"]; ok {
		outFay = v.(string)
	}
	PlanContext.topicEndDelete(getCheckMod(), dTopic)
	if strings.EqualFold(outFay, "bar") {
		PlanContext.TaskBarEndTopicDel(topicMsg)
	}
}
